package com.atguigu.flink.chapter11.function;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

public class Flink02_Table_error {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(2);

           DataStreamSource<String> waterSensorStream =
                   env.fromElements( "hello world","hello hello","hello flink11 flink22");

           // 1. 创建表的执行环境
           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
           Table table = tableEnv.fromDataStream(waterSensorStream,$("line"));
           tableEnv.createTemporaryView("s",table);

           //1.在table API 中使用
           // 1.1 内联的方式使用
           /*table
                //   .joinLateral(call(MySplit.class,$("line")))     // 内连接
                   .leftOuterJoinLateral(call(MySplit.class,$("line")))  // 左外连接
                   .select($("line"),$("word"),$("len"))
                   .execute()
                   .print();
           */
           // 1.2 函数先注册，再使用



           // 2.在sql 中使用
           // 先注册函数
           tableEnv.createTemporaryFunction("my_split",MySplit.class);
          // 只能使用外部类，不能使用WordLen 的方法，只能使用ROW
           tableEnv.sqlQuery("select " +
                   " line," +
                   " word," +
                   " len " +
                   "from s " +
                   "join lateral table(my_split(line)) on true")
                   .execute()
                   .print(); //内连接方式
           // select ...from a join b on a.id = b.id
           // select ...from a , b where a.id = b.id
           /*   tableEnv.sqlQuery("select " +
                   " line," +
                   " word," +
                   " len " +
                   "from s " +
                   ", lateral table(my_split(line)) ")
                   .execute()
                   .print();
*/
           // 外连接
          /* tableEnv.sqlQuery("select " +
                          " line," +
                          " word," +
                          " len " +
                          "from s " +
                          "left join lateral table(my_split(line)) on true ")
            .execute()
            .print();

*/




       }

          /* @FunctionHint(output = @DataTypeHint("row<word string,len int>"))
           public static class MySplit extends TableFunction<Row> {
                // 方法名和修饰符必须这样写，其他根据具体需求
           public void eval(String line){
               for (String word : line.split(" ")) {

                   if (line.contains("flink11")) {
                       return;
                   }
                   collect(Row.of(word,word.length()));
               }
               }

           }*/

    public static class MySplit extends TableFunction<WordLen> {
        // 方法名和修饰符必须这样写，其他根据具体需求
        public void eval(String line){
            for (String word : line.split(" ")) {
                collect(new WordLen(word,word.length()));
            }
        }

    }
                //  警示@#￥￥%#￥@#￥@#￥@#￥
                // 建议多用包装类型！！ 基本类型必须有值！！

    public static  class WordLen{
        public String word;
        // 包装类会更好，基本类会报错
        // 主要是针对内连接，而外连接不受影响！！！
        // 错误 CHARACTER SET "UTF-16LE" line, VARCHAR(2147483647)
        // CHARACTER SET "UTF-16LE" word, INTEGER NOT NULL len) NOT NULL
        public Integer len;
       // public int len;
        // 后面的int 和  Integer 不受影响，主要是成员变量的 类型要是包装类！！

        public WordLen() {

        }
        public WordLen(String word, int len) {
            this.word = word;
            this.len = len;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public int getLen() {
            return len;
        }

        public void setLen(int len) {
            this.len = len;
        }
    }


}


